package com.kafka.consumer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * author: LCG
 * Date: 2022-04-14 22:34:03
 * description: 获取消息的操作方式之一
 **/
@Slf4j
//@Component
public class KafKaListener {

    @KafkaListener(topics = {"TEST-TOPIC"},groupId = "consumer1",concurrency = "2")
    public void handler(ConsumerRecord<String,String> consumerRecord) throws InterruptedException {

        long offset = consumerRecord.offset();
        int partition = consumerRecord.partition();
        Object message = consumerRecord.value();

        System.out.println("====offset==>"+offset+"===分区====》"+partition+"====消息===》"+message);
        //System.out.println("====ALL=====>"+ JSON.toJSONString(consumerRecord));
        //Thread.sleep(10000);

    }

}
